package com.arica.mq.consumer.processor;

import com.arica.framework.dto.MqConfigDTO;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Service;

import java.util.*;

@Service
public class ProcessorFactory implements ApplicationListener<ContextRefreshedEvent>, ApplicationContextAware {
    private static final Logger LOGGER = LoggerFactory.getLogger(ProcessorFactory.class);

    private ApplicationContext context;

    private static Map<String, List<String>> topicTagMap = new HashMap<>();
    private static Map<String, List<MessageProcessor>> processorMap = new HashMap<>();

    public static List<MessageProcessor> getProcessors(String topic, String tag) {
        if (StringUtils.isBlank(topic) || StringUtils.isBlank(tag)) {
            return null;
        }

        return processorMap.get(topic + tag);
    }

    public String getTagsByTopic(String topic) {
        if (CollectionUtils.isEmpty(topicTagMap.get(topic))) {
            return null;
        }

        String tags = "";
        List<String> tagList = topicTagMap.get(topic);
        for (int i = 0; i < tagList.size(); i++) {
            if (i == tagList.size() - 1) {
                tags += tagList.get(i);
            } else {
                tags = tags + tagList.get(i) + "||";
            }
        }

        return tags;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.context = applicationContext;
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        if (event.getApplicationContext().getParent() != null) {
            return;
        }

        // 扫描出所有已经装配的 MessageProcessor 类 bean
        Map<String, MessageProcessor> processes = context.getBeansOfType(MessageProcessor.class);

        Set<String> processorKeySet = new HashSet<>();
        for (MessageProcessor process : processes.values()) {
            LOGGER.info(process.getProcessType() + " be created");

            // processType 不允许为空，也不允许重复
            String processType = process.getProcessType();
            if (StringUtils.isBlank(processType)) {
                throw new ClassFormatError(process.getClass() + ":processType can not null");
            }
            if (processorKeySet.contains(processType)) {
                throw new ClassFormatError("DuplicateKey processType = '" + processType + "'");
            }
            processorKeySet.add(processType);

            // mqConfig 为空则不处理
            List<MqConfigDTO> mqTagConfigList = process.getMqConfig();
            if (CollectionUtils.isEmpty(mqTagConfigList)) {
                continue;
            }

            for (MqConfigDTO mqTagConfig : mqTagConfigList) {
                String topic = mqTagConfig.getTopic();
                String tag = mqTagConfig.getTag();
                String key = topic + tag;

                // 添加 processor
                List<MessageProcessor> processorList = processorMap.computeIfAbsent(key, k -> new ArrayList<>());
                processorList.add(process);

                // 添加 tag
                if (CollectionUtils.isEmpty(topicTagMap.get(topic))) {
                    List<String> tags = new ArrayList<>();
                    tags.add(tag);
                    topicTagMap.put(topic, tags);
                } else {
                    if (!topicTagMap.get(topic).contains(tag)) {
                        topicTagMap.get(topic).add(tag);
                    }
                }
            }
        }
    }

}
